package com.wtw.sparkstreaming

import java.util
import java.util.concurrent.{Executors, TimeUnit}
import scala.collection.JavaConverters._

import com.wtw.utils.RedisUtil


/**
  * 王庭伟201731102227
  */
object KafkaSparkStreamingTimer {
  def main(args: Array[String]): Unit = {

    //启动一个定时任务 每隔10分钟来消费redis缓存中没有消费到的数据
    val service = Executors.newSingleThreadScheduledExecutor()

    service.scheduleAtFixedRate(new Runnable {
      override def run(): Unit = {
        //获取redis中的所有键值
        val jedis = RedisUtil.getConnection()
        val res: util.Set[String] = jedis.keys("spark-kafka_*")
        val keys = res.asScala
        keys.foreach(key => {
          val ttl = jedis.ttl(key)
          //判断存活超过十分钟就将其处理
          if(ttl == -1){
            //此处应该有校验逻辑 但是我不会
            //消费数据
            println(jedis.get(key))

            //销毁数据
            jedis.expire(key, 5)
          }
        })

        RedisUtil.returnConnection(jedis)
      }
    }, 10, 10, TimeUnit.MINUTES)

  }
}
